package com.ferry.taskflow.taskFlow.tx;

import com.ferry.taskflow.taskFlow.base.IdAble;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

/**
 * Created by daiyong
 * 事务协调器
 */
@Slf4j
public class TxCoordinater extends IdAble {

	private TxFlow txFlow;

	/**
	 * 检查事务线程
	 */
	private final ExecutorService watchExecutor;


	/**
	 * 期望的事务数量
	 */
	private AtomicInteger expectedTxCount = new AtomicInteger(0);

	/**
	 * 启动的事务数量
	 */
	private AtomicInteger startedTxCount = new AtomicInteger(0);

	/**
	 * 就绪的事务数量
	 */
	private AtomicInteger readyTxCount = new AtomicInteger(0);

	/**
	 * 回滚的事务数量
	 */
	private AtomicInteger rollbackTxCount = new AtomicInteger(0);

	/**
	 * 事务处理过程中是否已异常
	 */
	private AtomicBoolean isException = new AtomicBoolean(false);

	/**
	 * 是否已经关闭
	 */
	private final AtomicBoolean shutdown = new AtomicBoolean(false);

	/**
	 * 是否已经释放线程
	 */
	private final AtomicBoolean release = new AtomicBoolean(false);

	/**
	 * 已经就绪的事务线程
	 */
	private List<Thread> readyThreads = Lists.newCopyOnWriteArrayList();

	/**
	 * 超时时间单位ms
	 */
	private static final Long TIMEOUT = 10000L;

	public TxCoordinater(TxFlow txFlow) {
		this.txFlow = txFlow;

		watchExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("TxCoordinater-Watch"));

		startWatcher();
	}

	public void startWatcher() {
		watchExecutor.execute(() -> {
			long startTime = System.currentTimeMillis();

			//延迟5s执行
			try {
				TimeUnit.SECONDS.sleep(5);

				while (!shutdown.get()) {
					TimeUnit.MILLISECONDS.sleep(500);

					//如果超时 并且事务还没有就绪
					if ( (System.currentTimeMillis() - startTime) > TIMEOUT  && !decision()) {
						log.warn("TxCoordinater.watch timeout, flowID:{}, expectedTxCount:{}, readyTxCount:{}, rollbackTxCount:{}",
								txFlow.id(), expectedTxCount.get(), readyTxCount.get(), rollbackTxCount.get());
						markException();
						unParkTxThreads();
					}
				}
			} catch (InterruptedException e) {
				log.warn("TxCoordinater.watch Exception, flowID:{}", txFlow.id(), e);
			}
		});
	}

	/**
	 * 关闭协调器
	 */
	public void shutdown() {

		shutdown.getAndSet(true);

		if (null != watchExecutor) {
			watchExecutor.shutdown();
		}

		log.info("TxCoordinater shutdown now, flotID:{}", txFlow.id());
	}

	public void expect(int count) {
		int currExpectedTxCount = expectedTxCount.addAndGet(count);
		log.info("TxCoordinater.expect, flowID:{}, count:{}, currExpectedTxCount:{}", txFlow.id(), count, currExpectedTxCount);
	}

	public void start() {
		int currStartedTxCount = startedTxCount.incrementAndGet();
		log.info("TxCoordinater.start, flowID:{}, currStartedTxCount:{}", txFlow.id(), currStartedTxCount);
	}

	public void successReady() {
		Thread thread = Thread.currentThread();
		readyThreads.add(thread);
		int currReadyTxCount = readyTxCount.incrementAndGet();
		log.info("TxCoordinater.successReady, flowID:{}, thread:{}, currReadyTxCount:{}", txFlow.id(), thread, currReadyTxCount);

		decisionUnPark();

		//阻塞当前线程
		LockSupport.park();
	}

	/**
	 * 失败就绪只需要添加异常标记，executor会处理事务的回滚
	 */
	public void failReady() {
		Thread thread = Thread.currentThread();
		int currRollbackTxCount = rollbackTxCount.incrementAndGet();
		markException();
		log.info("TxCoordinater.failReady, flowID:{}, thread:{}, currRollbackTxCount:{}", txFlow.id(), thread, currRollbackTxCount);

		decisionUnPark();
	}

	private void decisionUnPark() {
		if (decision()) {
			unParkTxThreads();
		}
	}

	/**
	 * 标记异常
	 */
	private void markException() {
		isException.set(true);
	}

	/**
	 * 放行线程
	 */
	private synchronized void unParkTxThreads() {
		if (!release.get()) {
			for (Thread readyThread : readyThreads) {
				LockSupport.unpark(readyThread);
				log.info("TxCoordinater.unParkTxThreads, flowID:{}, thread:{}", txFlow.id(), readyThread);
			}
			release.getAndSet(true);
			shutdown();
		}
	}

	private boolean decision() {
		return expectedTxCount.get() == (readyTxCount.get() + rollbackTxCount.get());
	}


	public boolean shouldRollback() {
		return isException.get();
	}
}
